package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.message.ArticleVisitStreamMessage;
import com.heima.model.message.UpdateArticleMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Slf4j
@Configuration
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        // 接收消息
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        // 聚合流式处理
        kStream.map((key, value) -> {
                    UpdateArticleMessage message = JSON.parseObject(value, UpdateArticleMessage.class);
                    // 重置消息的key和value
                    return new KeyValue<>(message.getArticleId().toString(), message.getType().name() + ":" + message.getAdd());
                })
                // 按照时间id进行聚合
                .groupBy((key, value) -> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                // 自行完成聚合的计算
                .aggregate(new Initializer<String>() {
                    // 初始方法，返回值是消息的value
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        if (StringUtils.isBlank(value)) {
                            return aggValue;
                        }
                        String[] aggArr = aggValue.split(",");
                        int collection = 0, comment = 0, likes = 0, views = 0;
                        for (String agg : aggArr) {
                            String[] split = agg.split(":");
                            /**
                             * 获取初始值，也是时间窗口内计算之后的值
                             */
                            switch (UpdateArticleMessage.UpdateArticleType.valueOf(split[0])) {
                                case COLLECTION:
                                    collection = Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    comment = Integer.parseInt(split[1]);
                                    break;
                                case LIKES:
                                    likes = Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    views = Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        /**
                         * 累加操作
                         */
                        String[] valArr = value.split(":");
                        switch (UpdateArticleMessage.UpdateArticleType.valueOf(valArr[0])) {
                            case COLLECTION:
                                collection += Integer.parseInt(valArr[1]);
                                break;
                            case COMMENT:
                                comment += Integer.parseInt(valArr[1]);
                                break;
                            case LIKES:
                                likes += Integer.parseInt(valArr[1]);
                                break;
                            case VIEWS:
                                views += Integer.parseInt(valArr[1]);
                                break;
                        }
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, likes, views);
                        log.info("文章的id：" + key);
                        log.info("当前时间窗口内的消息处理结果：" + formatStr);
                        System.out.println("文章的id：" + key);
                        System.out.println("当前时间窗口内的消息处理结果：" + formatStr);
                        return formatStr;
                    }
                }, Materialized.as("hot-article-count-001")) // 流式处理的状态
                .toStream()
                .map((key, value) -> new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value)))
                // 发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return kStream;
    }

    /**
     * 格式化消息的value数据
     *
     * @param articleId
     * @param value
     * @return
     */
    private String formatObj(String articleId, String value) {
        ArticleVisitStreamMessage articleVisitStreamMessage = new ArticleVisitStreamMessage();
        articleVisitStreamMessage.setArticleId(Long.valueOf(articleId));
        // "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
        String[] valueArr = value.split(",");
        for (String val : valueArr) {
            String[] split = val.split(":");
            switch (UpdateArticleMessage.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    articleVisitStreamMessage.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    articleVisitStreamMessage.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    articleVisitStreamMessage.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    articleVisitStreamMessage.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为：" + JSON.toJSONString(articleVisitStreamMessage));
        return JSON.toJSONString(articleVisitStreamMessage);
    }
}
